package com.flink.flinkdemo.demo;

import org.apache.flink.api.common.functions.CoGroupFunction;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.operators.CoGroupOperator;
import org.apache.flink.api.java.operators.DataSource;
import org.apache.flink.api.java.operators.FlatMapOperator;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.util.Collector;

/**
 * @author zhangy on 2020/12/31
 * @description: add you description
 */
public class CoGroupDemo {

    public static void main(String[] args) throws Exception {


        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
        DataSource<String> coG1 = env.readTextFile("/Users/zhangyong/Idea-git/flink-d/flink-demo/src/doc/CoG1");
        FlatMapOperator<String, Tuple2<String, Integer>> coG1FlatMap = coG1.flatMap(new FlatMapFunction<String, Tuple2<String, Integer>>() {

            @Override
            public void flatMap(String value, Collector<Tuple2<String, Integer>> out) throws Exception {
                String[] words = value.split(" ");
                out.collect(new Tuple2<String, Integer>(words[0], Integer.valueOf(words[1])));

            }
        });


        DataSource<String> coG2 = env.readTextFile("/Users/zhangyong/Idea-git/flink-d/flink-demo/src/doc/CoG2");

        FlatMapOperator<String, Tuple2<String, Integer>> coG2Flatmap = coG2.flatMap(new FlatMapFunction<String, Tuple2<String, Integer>>() {
            @Override
            public void flatMap(String value, Collector<Tuple2<String, Integer>> out) throws Exception {
                String[] s = value.split(" ");
                out.collect(new Tuple2<>(s[0], Integer.valueOf(s[1])));
            }
        });


        CoGroupOperator<Tuple2<String, Integer>, Tuple2<String, Integer>, Tuple3<String, Integer, Integer>> with = coG1FlatMap.coGroup(coG2Flatmap).where(0).equalTo(0).with(new CoGroupFunction<Tuple2<String, Integer>, Tuple2<String, Integer>, Tuple3<String, Integer, Integer>>() {
            @Override
            public void coGroup(Iterable<Tuple2<String, Integer>> first, Iterable<Tuple2<String, Integer>> second, Collector<Tuple3<String, Integer, Integer>> out) throws Exception {

                for (Tuple2<String, Integer> f : first) {
                    for (Tuple2<String, Integer> s : second) {
                        if (f.f0.equals(s.f0)) {
                            out.collect(new Tuple3<>(f.f0, f.f1, s.f1));
                        }
                    }
                }
            }
        });
        with.print();
    }
}